package org.example.flume.source;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class MySource extends AbstractSource implements Configurable, PollableSource {
    private Status status = Status.READY;
    private String prefix = "";
    @Override
    public Status process() throws EventDeliveryException {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        try {
            Event event = getSomeData();
            getChannelProcessor().processEvent(event);
            status = Status.READY;
        } catch (Exception e) {
            status = Status.BACKOFF;
            e.printStackTrace();
        }
        return status;
    }

    private Event getSomeData() {
        String str = UUID.randomUUID().toString();
        Event event = new SimpleEvent();
        event.setBody(str.getBytes());
        return event;
    }

    @Override
    public long getBackOffSleepIncrement() {
        return 1000;
    }

    @Override
    public long getMaxBackOffSleepInterval() {
        return 10000;
    }

    @Override
    public void configure(Context context) {
        prefix = context.getString("prefix", "logs-");
    }
}
